package com.study.flink.table

import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.table.api.TableEnvironment
import org.apache.flink.types.Row

/**
  * Table & SQL
  *
  * @author: stephen.shen
  * @create: 2019-05-27 16:15
  */
object FlinkTableSQLDemo {

  def main(args: Array[String]): Unit = {
    // 1、执行环境
    val env = ExecutionEnvironment.getExecutionEnvironment
    val tableEnv = TableEnvironment.getTableEnvironment(env)

    // 2、指定数据源
    val filePath = "xxx.csv"
    import org.apache.flink.api.scala._
    val csv = env.readCsvFile[SalesLog](filePath, ignoreFirstLine = true)

    // DataSet => Table
    val salesTable = tableEnv.fromDataSet(csv)

    // 注册成一张表
    tableEnv.registerTable("sales", salesTable)

    // 使用SQL进行操作
    val resultTable = tableEnv.sqlQuery("SELECT customerId, SUM(amountPaid) money FROM sales GROUP BY customerId")

    // 转换成DataSet
    tableEnv.toDataSet[Row](resultTable).print()

    // 启动任务
    env.execute("Table & SQL Demo")
  }

  case class SalesLog(transactionId: String,
                      customerId: String,
                      itemId: String,
                      amountPaid: Double)

}
